/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.exception.ExceptionUtils;
import org.apache.commons.lang3.StringUtils;
import org.ipig.commons.helper.AssertHelper;
import org.ipig.commons.helper.DateTimeHelper;
import org.ipig.computing.constant.context.SparkContext;
import org.ipig.computing.spark.conf.ApplicationConf;
import org.ipig.computing.spark.conf.SparkLauncherConf;
import org.ipig.constants.ResultStatus;
import org.ipig.messaging.ProducerService;
import org.ipig.messaging.kafka.producer.KafkaProducerService;

import java.lang.reflect.Method;
import java.text.MessageFormat;

/**
 * AbstractSparkLauncher
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
@Slf4j
public abstract class AbstractSparkLauncher implements SparkLauncherService<SparkLauncherConf> {
    /**
     * 获取生产者服务
     * @return
     */
    public ProducerService getProducerService(){
        throw new IllegalArgumentException("请先注入生产者服务bean");
    }

    @Override
    public void setConf(SparkLauncherConf sparkLauncherConf) {

    }

    @Override
    public ResultStatus launch() {
        init();
        String mainArg = getMainArg();
        String mainClass = getMainClass();
        String appName = this.getAppName();
        AssertHelper.hasText(mainClass, MessageFormat.format("【{0}】  请设置application的 mainClass",getConf().getAppName()));
        AssertHelper.hasText(mainArg, MessageFormat.format("【{0}】  请设置application的 mainArg",getConf().getAppName()));
        AssertHelper.hasText(appName, MessageFormat.format("【{0}】  请设置application的全局唯一的appName,",getConf().getAppName()));
        ResultStatus result = ResultStatus.SUCCESS;
        String master = getConf().getMaster();
        String workingMode = SparkContext.WorkingMode.YARN.code;
        String[] appArgs = new String[]{mainArg, workingMode};
        try {
            beforeComputing();
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(ExceptionUtils.getFullStackTrace(e));
        }

        if (StringUtils.startsWithIgnoreCase(master, SparkContext.WorkingMode.LOCAL.code)) {
            workingMode = SparkContext.WorkingMode.LOCAL.code;
            appArgs[1] = workingMode;
            Method mainMethod;
            try {
                log.warn("开始处理应用【{}】,{}",appName ,SparkContext.WorkingMode.LOCAL.title);
                mainMethod = Class.forName(mainClass).getMethod(SparkContext.MAIN_FUNCTION, String[].class);
                mainMethod.invoke(null, (Object) appArgs);
            } catch (Exception e) {
                result = ResultStatus.FAILURE;
                e.printStackTrace();
                System.out.println(ExceptionUtils.getFullStackTrace(e));
            }
        } else {
            ApplicationConf app = new ApplicationConf();
            app.setAppName(appName);
            app.setMainClass(mainClass);
            app.setMainArg(mainArg);
            app.setAppLibDir(getConf().getAppLibDir());
            app.setAppResource(getConf().getAppResource());
            app.setAppDeployMode(getConf().getAppDeployMode());
            app.setCreateTime(DateTimeHelper.getDateTime());
            try {
                log.warn("开始初始应用【{}】,{}",appName , SparkContext.WorkingMode.YARN.title);
                String json = app.toJson();
                log.debug(json);
                if (getProducerService() != null) {
                    if (getProducerService() instanceof KafkaProducerService) {
                        KafkaProducerService producerService = (KafkaProducerService) getProducerService();
                        producerService.send(json);
                    } else {
                        throw new IllegalArgumentException(MessageFormat.format("不识ProducerService: 【{0}】",getProducerService().getClass()));
                    }
                } else {
                    throw new IllegalArgumentException("没有ProducerService可用");
                }
            } catch (Exception e) {
                result = ResultStatus.FAILURE;
                e.printStackTrace();
                log.error(ExceptionUtils.getFullStackTrace(e));
            }
        }

        try {
            afterComputing();
        } catch (Exception e) {
            e.printStackTrace();
            log.error(ExceptionUtils.getFullStackTrace(e));
        }
        if (StringUtils.equals(workingMode,SparkContext.WorkingMode.LOCAL.code) ) {
            log.warn("完成执行应用【{}】,{}",appName ,SparkContext.WorkingMode.LOCAL.title);
        }else{
            log.warn("完成初始应用【{}】,{}，应用将在集群中运行。",appName , SparkContext.WorkingMode.YARN.title);
        }
        return result;
    }
}
